package bch.rocketmq.service;

import lombok.extern.slf4j.Slf4j;
import org.springframework.stereotype.Component;

import java.util.Arrays;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.locks.Condition;
import java.util.concurrent.locks.Lock;
import java.util.concurrent.locks.ReentrantLock;

/**
 * 有界容器
 */
@Slf4j
@Component
public class BoundedContainer {

    // 元素
    private String[] elements = new String[10];

    // 操作交互安全性保证
    private Lock lock = new ReentrantLock();

    /**
     * 虚假唤醒：线程被唤醒后，还需在循环中重新检查条件（while 而不是 if）
     * 而aqs的多条件队列避免了虚假唤醒，但仍需在循环中检查条件。
     */
    // 有界数组 满
    private Condition notEmpty = lock.newCondition();

    // 有界数组 空
    private Condition notFull = lock.newCondition();

    // region int常量 不会有安全问题
    // 实际元素数量
    private int elementCount;

    // 放置元素索引
    private int putIndex;

    // 提取元素索引
    private int takeIndex;

    // endregion


    public void put(String ele) throws InterruptedException {
        lock.lock();

        try {
            // 如果容器已满, 等待
            while (this.elementCount == elements.length) {
                // 调用 await() 时：原子释放锁 → 线程不再持有锁
                // 线程被唤醒后，会先加入 同步队列（Sync Queue），排队等待锁,拿到锁后，await() 才返回
                // 被唤醒线程在 Condition 队列中等待时，锁已经被释放。唤醒后必须重新竞争锁，所以它不保证立刻进入临界区，公平性决定了竞争的顺序
                // 可以使用带超时时间的await() 方法，避免长时间阻塞多个线程, 像ArrayBlockingQueue提供一些带超时参数的方法
                notFull.await();
            }

            // 放置元素
            elements[putIndex] = ele;

            // 放置元素索引自增
            if (++putIndex == this.elements.length) {
                // 循环链表
                putIndex = 0;
            }

            // 元素数量
            ++elementCount;

            log.info("Put:{}", Arrays.toString(elements));

            // 通知不为空
            notEmpty.signal();

        } finally {
            lock.unlock();
        }

    }

    public Object take() throws InterruptedException {
        lock.lock();

        try {
            // 如果容器为空,等待
            while (elementCount == 0) {
                boolean await = notEmpty.await(2, TimeUnit.SECONDS);
                if (!await) {
                    log.info("等待超时");
                    return null;
                }
            }


            Object o = elements[takeIndex];

            elements[takeIndex] = null;


            // 取值元素索引自增
            if (++takeIndex == this.elements.length) {
                // 循环链表
                takeIndex = 0;
            }

            // 元素数量
            --elementCount;

            log.info("Take:{}", Arrays.toString(elements));

            notFull.signal();

            return o;
        } finally {
            lock.unlock();
        }

    }

    @Override
    public String toString() {
        return "BoundedContainer{" +
                "elements=" + Arrays.toString(elements) +
                '}';
    }
}
